package app_kvServer;

import infrastructure.HasherService;
import infrastructure.KVMessageImp;
import infrastructure.LoggingManager;
import java.security.NoSuchAlgorithmException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;

import clientCommunication.CommunicationFactory;
import clientCommunication.CommunicationSync;
import clientCommunication.ICommunication;

import common.messages.KVMessage;
import common.messages.KVMessage.MessageType;
import common.messages.KVMessage.StatusType;

import serverCommunication.CommunicationListener;
import serverCommunication.IResponse;
import serverCommunication.Server;


public class KVServer implements CommunicationListener {
	private Logger logger = LoggingManager.CreateServerLogger(KVServer.class);
	private Server server;
	private ServerStatus statusServer;
	Map<String, String> keyValueStorage;
	Map<String, String> metadata;
	private String myServerInfo = "";
	private HasherService hasherService;
	private String myKeysMetadata;
	private String myValuesMetadata;
	
	
	/**
	 * Start KV Server at given port
	 * @param port given port for storage server to operate
	 * @throws NoSuchAlgorithmException 
	 */
	public KVServer(int port) {
		this.keyValueStorage = new HashMap<String, String>();
		this.metadata = new HashMap<String, String>();
		this.statusServer = ServerStatus.STOPPED;
		try {
			this.hasherService = new HasherService();
		} catch (NoSuchAlgorithmException e) {
			logger.error("Error creating hasher service ", e);
		}
		server = new Server(port, this);
		server.start();		
		System.out.println( "Running in port " + port);
	}
	
	public KVServer(int port, String key, String value, String to) {
		this(port);
		this.fillMetadata(key, value, to);
		this.statusServer = ServerStatus.STARTED;
	}
	
	/**
     * Main entry point for the echo server application. 
     * @param args contains the port number at args[0].
	 * @throws NoSuchAlgorithmException 
     */
    public static void main(String[] args) throws NoSuchAlgorithmException {
    	try {
    		int port = 50000;    	
			if(args.length != 1) {
				System.out.println("Using default port");
			}
			else {
				port = Integer.parseInt(args[0]);
			}
			
			new KVServer(port);		
		} catch (NumberFormatException nfe) {
			System.out.println("Error! Invalid argument <port>! Not a number!");
			System.out.println("Usage: Server <port>!");
			System.exit(1);
		}
    }

	@Override
	public void notifyMessage(KVMessage message, IResponse response) {
		// Process Message
		KVMessage result = null;
		String replyMessage = "";
		try {
			result = handleMessage(message);
		} catch (Exception e1) {
			replyMessage = "ERROR in notify message method: " + e1.getMessage();
			logger.error(replyMessage, e1);
		}
		
		if (result != null) {
			response.sendResponse(result);
		}
		
		 if (message.getStatus().equals(StatusType.SHUTDOWN)){
			 this.server.stopServer();
		 }
	}

	private synchronized KVMessage handleMessage(KVMessage message) throws Exception {
		KVMessage result = null; 
		
		if (message.getMessageType().equals(MessageType.KV)){
			logger.info("handling KV message");
			result = handleKvMessage(message);
			
		} else if (message.getMessageType().equals(MessageType.METADATA)) {
			logger.info("handling metadata message");
			result = handleMetadataMessage(message); 
		}
		
		return result;
	}

	private KVMessage handleMetadataMessage(KVMessage message) {
		KVMessage result = null; 
		String key, value;
		
		if (message.getStatus().equals(StatusType.START)){
			this.statusServer = ServerStatus.LOCKED;
			this.fillMetadata(message.getKey(), message.getValue(), message.getTo());
			this.reallocateData();
			this.statusServer = ServerStatus.STARTED;
		} else if (message.getStatus().equals(StatusType.STOP)){
			this.statusServer = ServerStatus.STOPPED;
		}  else if (message.getStatus().equals(StatusType.PUT_ALL)){
			key = message.getKey();
			value = message.getValue();
			if (key != null && value != null){
				String [] keys = key.split(KVMessage.STRING_SEPARATOR);
				String [] values = value.split(KVMessage.STRING_SEPARATOR);
				for(int i =0 ; i < keys.length; i ++){
					this.keyValueStorage.put(keys[i], values[i]);
				}
			}
		} else if (message.getStatus().equals(StatusType.SHUTDOWN)){
			this.fillMetadata(message.getKey(), message.getValue(), message.getTo());
			this.reallocateData();
			// Response any reply to confirm that this server can be take down already
			result = new KVMessageImp(MessageType.METADATA, StatusType.SHUTDOWN);
		
		}
		return result;
	}

	private KVMessage handleKvMessage(KVMessage message) {
		KVMessage result = null;
		String key, value = null;
		StatusType statusMessage;
		
		if (this.statusServer.equals(ServerStatus.STOPPED)) {
			result = new KVMessageImp(MessageType.KV, StatusType.SERVER_STOPPED);
		} else if (this.statusServer.equals(ServerStatus.LOCKED)) {
			result = new KVMessageImp(MessageType.KV, StatusType.SERVER_WRITE_LOCK);
		}
		else {
			if (!this.validateMessage(message)){
				logger.info("Message key does not correspond to this server " + this.myServerInfo);
				result = new KVMessageImp(MessageType.METADATA, this.myValuesMetadata, this.myKeysMetadata, StatusType.SERVER_NOT_RESPONSIBLE);	
			} else {
				logger.info("Message corresponds to this server " + this.myServerInfo);
				if (message.getStatus().equals(StatusType.GET)) {
					key = message.getKey();
					statusMessage = StatusType.GET;			
					if ( this.keyValueStorage.containsKey(key)) {
						statusMessage = StatusType.GET_SUCCESS;
						value = this.keyValueStorage.get(key);
					}
					else {
						statusMessage = StatusType.GET_ERROR;
						value = "";
					}
					
					result = new KVMessageImp(MessageType.KV, value, key, statusMessage);
				} else if (message.getStatus().equals(StatusType.PUT)) {
					key = message.getKey();
					statusMessage = StatusType.PUT;
					value = message.getValue();
					if (value == null || value.equals("") || value.equals("null")) {
						statusMessage = StatusType.DELETE_ERROR;
						if (this.keyValueStorage.containsKey(key)){
							this.keyValueStorage.remove(key);
							statusMessage = StatusType.DELETE_SUCCESS;
						}
					}
					else {
						if (this.keyValueStorage.containsKey(key)){
							statusMessage = StatusType.PUT_UPDATE;	
							
						} else {
							statusMessage = StatusType.PUT_SUCCESS;	
						}
						
						this.keyValueStorage.put(key, value);
					}
					
					result = new KVMessageImp(MessageType.KV, value, key, statusMessage);
				}
			}
		}
		
		return result;
	}	
	

	/**
	 * Check that the message is handled by this server. (HASH SET)
	 * @param message
	 * @return
	 */
	private boolean validateMessage(KVMessage message) {
		// TODO: This implementation is not the optimal. Since it would be better to keep the range
		// of the hashes. Here we only have one value and then we have loop through and rely on the order
		// of the collection.
		
		logger.info("Validating Message " + message.getKey() + " / " + message.getValue());
		String hashKeyData = this.hasherService.getHash(message.getKey());		
		logger.info(String.format("Hashes to compare %s and %s", hashKeyData, hasherService.getUpperHashBound()));
		return this.hasherService.isHashInRange(hashKeyData);//, this.myLowerHashBound, this.myUpperHashBound);
	}

	private void reallocateData() {
		if (this.metadata.isEmpty())
			return;
		logger.info("Reallocating data ... ");
		Map<String, KVMessageImp> dataToReallocate = new HashMap();
		KVMessageImp message;
		for(String key:this.keyValueStorage.keySet()){
			//String hashKeyData = this.hasherService.getHash(key);
			String serverInfo = this.hasherService.getServerInfo(this.metadata, key);
			if (serverInfo != null) {
				// If the key is meant to be distributed to the same server, so do not distribute.
				if (serverInfo != this.myServerInfo) {
					if (!dataToReallocate.containsKey(serverInfo)){
						message = new KVMessageImp(MessageType.METADATA, "", "", StatusType.PUT_ALL);
						message.setTo(serverInfo);						
						dataToReallocate.put(serverInfo, message);							
					}
					
					message = dataToReallocate.get(serverInfo);
					String currentKey = message.getKey();
					String currentValue = message.getValue();
					currentKey += key + ";";
					currentValue += this.keyValueStorage.get(key) + ";";
					message.setKey(currentKey);
					message.setValue(currentValue);
				}				
			} else {
				logger.warn("Server was not found to distribute the key " + key);
			}
				
		}
		
		
		sendPutAllMessage(dataToReallocate.values());		
	}

	private void sendPutAllMessage(Collection<KVMessageImp> collection) {
		for (KVMessageImp message:collection){
			String[] tokens = message.getTo().split(":");
			String ipaddress = tokens[0];
			String port = tokens[1];
			logger.info(String.format("Distributing data from %s to %s", this.myServerInfo, message.getTo()));
			CommunicationSync sync;
			try {
				ICommunication com = CommunicationFactory.createCommunication(port, ipaddress, true);
				com.run();
				logger. info(com.receiveMessage().getValue());
				com.sendMessage(message);
				com.disconnect();
				
			} catch (Exception e) {
				logger.error("Error redistributing message ", e);
			}			
		}
	}

	private void fillMetadata(String key, String value, String to) {
		logger.info(String.format("Server setting metadata for %s. key : %s value: %s", to, key, value));		
		this.myServerInfo = to;		
		this.myKeysMetadata = key;
		this.myValuesMetadata = value;
		if (key == null || key.equals("")){
			this.metadata.clear();
			return;
		}
		// It is important to get the server info, so it uses it to calculate 
		// the higher and lower bound.
		this.hasherService.setMyServerInfo(this.myServerInfo);
		this.metadata = hasherService.getHashing(key, value);		
	}
}
